package org.btik.lightdev.dev.udp;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.btik.context.thread.SystemControlAble;
import org.btik.light.meta.DefaultDevMo;
import org.btik.light.meta.Dev;
import org.btik.light.meta.DevMo;
import org.btik.light.meta.DevState;
import org.btik.light.meta.base.CmdCode;
import org.btik.light.server.platform.common.api.bean.dev.access.DevAccount;
import org.btik.light.server.platform.common.api.dao.LightDao;
import org.btik.light.server.platform.common.api.service.DevAccessSever;
import org.btik.light.server.platform.common.timeout.TimeOutCache;
import org.btik.light.server.platform.manage.DevEventDispatcher;
import org.btik.light.tool.ByteUtil;
import org.btik.light.tool.MDUtil;
import org.btik.light.tool.UuidUtil;
import org.btik.light.tool.type.HashByteArray;
import org.btik.light.tool.type.HashByteArrayFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.*;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;

import static java.lang.Math.max;
import static org.btik.light.meta.LightDevAttrConstant.ACCESS_CHANNEL_TYPE;
import static org.btik.light.tool.ByteUtil.*;
import static org.btik.lightdev.dev.udp.ActionConstant.*;

/**
 * @author lustre
 * @since 2023/9/17 11:01
 */
public class UdpDevAccessServer extends Thread implements DevAccessSever, SystemControlAble {
    private final Log log = LogFactory.getLog(UdpDevAccessServer.class);

    private static final ConcurrentHashMap<String, UdpDev> SN_2_DEV = new ConcurrentHashMap<>();
    private volatile boolean runFlag = true;
    private DatagramSocket serverSocket;
    private int accessServerPort = 8081;

    private TimeOutCache timeOutCache;

    private DevEventDispatcher devEventDispatcher;

    private ReportHandler reportHandler;

    private int queueSize = 1024;

    private long nonceTimeout = 5 * 60 * 1000;

    private LightDao lightDao;

    public void setAccessServerPort(int accessServerPort) {
        this.accessServerPort = accessServerPort;
    }

    public void setDevEventDispatcher(DevEventDispatcher devEventDispatcher) {
        this.devEventDispatcher = devEventDispatcher;
    }

    public void setQueueSize(int queueSize) {
        this.queueSize = max(queueSize, 8);
    }

    public void setTimeOutCache(TimeOutCache timeOutCache) {
        this.timeOutCache = timeOutCache;
    }

    public void setNonceTimeout(long nonceTimeout) {
        this.nonceTimeout = nonceTimeout;
    }


    public void setLightDao(LightDao lightDao) {
        this.lightDao = lightDao;
    }

    @Override
    public synchronized void start() {
        reportHandler = new ReportHandler();
        reportHandler.messages = new ArrayBlockingQueue<>(queueSize);
        super.start();
        reportHandler.start();
    }

    @Override
    public void run() {
        try {
            serverSocket = new DatagramSocket(accessServerPort);
        } catch (SocketException e) {
            log.error("", e);
            return;
        }
        byte[] buffer = new byte[1500];
        DatagramPacket datagramPacket = new DatagramPacket(buffer, 0, buffer.length);
        while (runFlag) {
            try {
                serverSocket.receive(datagramPacket);
                int length = datagramPacket.getLength();
                InetAddress address = datagramPacket.getAddress();
                Report report = new Report();
                report.address = address;
                report.port = datagramPacket.getPort();
                report.data = Arrays.copyOf(buffer, length);
                reportHandler.messages.add(report);
            } catch (Exception e) {
                log.error("", e);
            }
        }
    }

    @Override
    public String getSNFromAddress(SocketAddress devAddress) {
        return null;
    }

    @Override
    public Dev getDev(String sn) {
        return SN_2_DEV.get(sn);
    }

    @Override
    public void shutdown(String msg) {
        if (serverSocket != null) {
            serverSocket.close();
        }
        runFlag = false;
        log.warn("UdpDevAccessServer exited");
    }


    static class Report {
        byte[] data;

        InetAddress address;

        int port;
    }

    /**
     * @return 是否无io异常
     */
    boolean sendResponse(Report report, byte[] pack) {
        DatagramPacket datagramPacket = new DatagramPacket(pack, pack.length);
        try {
            datagramPacket.setAddress(report.address);
            datagramPacket.setPort(report.port);
            serverSocket.send(datagramPacket);
        } catch (IOException e) {
            log.error("send failed", e);
            return false;
        }
        return true;
    }

    class ReportHandler extends Thread {
        ArrayBlockingQueue<Report> messages;
        HashMap<HashByteArray, BiConsumer<Integer, Report>> actionHandlers = new HashMap<>();

        private static final String NONCE_NAMESPACE = "LIGHT_UDP_NONCE";

        private void onLogin(int index, Report report) {
            HashByteArray sn = readItemByLength(report.data, index);
            if (sn.isEmpty()) {
                log.error("Invalid sn");
                return;
            }
            index += sn.length() + 1;
            HashByteArray userName = readItemByLength(report.data, index);
            if (userName.isEmpty()) {
                log.error("Invalid userName");
                return;
            }
            index += userName.length() + 1;
            HashByteArray md = readItemByLength(report.data, index);
            if (md.isEmpty()) {
                log.error("Invalid md");
                return;
            }
            DevAccount devAccount = lightDao.get(userName.toString(), DevAccount.class);
            if (devAccount == null || !devAccount.getStatus()) {
                log.error("user not exists :" + userName);
                if (log.isDebugEnabled()) {
                    log.debug("userName:" + userName);
                }
                return;
            }

            MDUtil.Algorithm mdAlgorithm = devAccount.getMdAlgorithm();

            byte[] nonce = timeOutCache.getObj(NONCE_NAMESPACE, sn);
            if (nonce == null) {
                log.error("Invalid nonce by sn:" + toHexString(sn));
            }
            // 利用服务端用户数据计算摘要，与设备摘要比较
            // 密码本身是账户加密码算出的哈希值。此处在之前加入nonce,便与设备一致
            HashByteArray serverMd = new HashByteArray(MDUtil.digest(mdAlgorithm, nonce, ByteUtil.hexStrToByteArray(devAccount.getPasswd())));
            if (log.isDebugEnabled()) {
                log.debug("nonce:" + ByteUtil.toHexString(nonce));
                log.debug("passwdHash:" + devAccount.getPasswd());
                log.debug("serverMd:" + ByteUtil.toHexString(serverMd));
            }
            byte[] pack;
            if (serverMd.equals(md)) {
                pack = LOGIN_SUCCESS;
                String snHex = toHexString(sn);
                UdpDev dev = new UdpDev(snHex, report.address,report.port, serverSocket);
                SN_2_DEV.put(snHex, dev);
            } else {
                log.error("login failed nonce by sn:" + toHexString(sn) + "/" + report.address + "/" + report.port);
                pack = LOGIN_AUTH_FAILED;
            }
            sendResponse(report, pack);
        }

        private void onNonce(int index, Report report) {
            HashByteArray sn = readItemByLength(report.data, index);
            if (sn == null) {
                return;
            }
            // 客户端收发很快 也会重试发送和接受 重试时不重置nonce
            byte[] nonce = timeOutCache.getObj(NONCE_NAMESPACE, sn);
            if(nonce == null){
               nonce = UuidUtil.uuid(NONCE_LENGTH);
            }
            // NONCE 动作头 动作长度 1字节 动作名称 NONCE长度
            int nonceHeaderLength = NONCE.length();
            byte[] pack = new byte[NONCE_LENGTH + nonceHeaderLength + 2];
            pack[0] = (byte) nonceHeaderLength;
            HashByteArrayFactory.copy(NONCE, pack, 0, 1, nonceHeaderLength);
            pack[nonceHeaderLength+1] = NONCE_LENGTH;
            System.arraycopy(nonce, 0, pack, nonceHeaderLength + 2, NONCE_LENGTH);

            if (sendResponse(report, pack)) {
                timeOutCache.save(NONCE_NAMESPACE, sn, nonce, nonceTimeout);
            } else {
                log.error("send nonce failed");
            }
        }

        private void onHello(int index, Report report) {
            HashByteArray sn = readItemByLength(report.data, index);
            if (sn.isEmpty()) {
                log.error("Invalid sn");
                return;
            }
            String snHex = toHexString(sn);
            if (SN_2_DEV.get(snHex) == null) {
                log.error("Invalid sn");
                return;
            }
            index += sn.length() + 1;
            int attrCount = readUnsignedShort(report.data, index);
            if (attrCount == -1) {
                log.error("Invalid attrCount");
                return;
            }
            index += 2; // 2字节attrCount
            Map<String, String> props = new HashMap<>();
            ByteArrayOutputStream byteBuffer = new ByteArrayOutputStream();
            for (int i = 0; i < attrCount; i++) {
                String key = readUntil(byteBuffer, report.data, index, CmdCode.COLON);
                index += key.length() + 1;
                String value = readUntil(byteBuffer, report.data, index, CmdCode.NEW_LINE);
                index += value.length() + 1;
                // 删除\r
                props.put(key, value);
            }
            props.put(ACCESS_CHANNEL_TYPE, AccessChannelType.UDP.name());
            DevMo devMo = new DefaultDevMo(snHex, DevState.CONNECTED);
            devMo.setExtendedAttributes(props);
            sendResponse(report, HELLO_ACK);
            devEventDispatcher.onConnect(devMo);
            timeOutCache.removeObj(NONCE_NAMESPACE, sn);
        }

        private void onKeep(int index, Report report) {
            HashByteArray sn = readItemByLength(report.data, index);
            if (sn.isEmpty()) {
                log.error("Invalid sn");
                return;
            }
            String snHex = toHexString(sn);
            UdpDev dev = SN_2_DEV.get(snHex);
            if (dev  == null) {
                return;
            }
            dev.updateAddr(report.address, report.port);
        }


        public ReportHandler() {
            super("ReportHandler");
            actionHandlers.put(NONCE, this::onNonce);
            actionHandlers.put(LOGIN, this::onLogin);
            actionHandlers.put(HELLO, this::onHello);
            actionHandlers.put(KEEP, this::onKeep);
        }

        @Override
        public void run() {
            while (runFlag) {
                try {
                    Report take = messages.take();
                    HashByteArray key = readItemByLength(take.data, 0);
                    BiConsumer<Integer, Report> handler = actionHandlers.get(key);
                    int index = key.length();
                    if (handler != null && ++index < take.data.length) {
                        handler.accept(index, take);
                    }
                } catch (Exception e) {
                    log.warn("", e);
                    if (!runFlag) {
                        log.warn("ReportHandler exit");
                        break;
                    }
                }

            }
        }
    }
}
